package org.apache.rocketmq.broker.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
import org.apache.rocketmq.common.AbortProcessException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageType;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageResponseHeader;
import org.apache.rocketmq.store.stats.BrokerStatsManager;

import java.net.SocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/25
 * @方法描述：
 */
public abstract class AbstractSendMessageProcessor  implements NettyRequestProcessor {

    protected static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
    protected static final Logger DLQ_LOG = LoggerFactory.getLogger(LoggerName.DLQ_LOGGER_NAME);

    //protected List<ConsumeMessageHook> consumeMessageHookList;
    protected final static int DLQ_NUMS_PER_GROUP = 1;
    protected final BrokerController brokerController;
    protected final Random random = new Random(System.currentTimeMillis());
    private List<SendMessageHook> sendMessageHookList;


    public AbstractSendMessageProcessor(final BrokerController brokerController) {
        this.brokerController = brokerController;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验生产者客户端发送过来的消息是否合法的方法，该方法的逻辑非常简单，我就不添加注释了，大家自己看看就行
     */
    protected RemotingCommand msgCheck(final ChannelHandlerContext ctx,
                                       final SendMessageRequestHeader requestHeader, final RemotingCommand request,
                                       final RemotingCommand response) {
        //校验当前broker节点是否具备写权限
        if (!PermName.isWriteable(this.brokerController.getBrokerConfig().getBrokerPermission())
                //如果当前broker节点没有写权限，并且当前要处理的消息的主题为顺序主题，就返回错误响应，表示没有权限处理消息
                && this.brokerController.getTopicConfigManager().isOrderTopic(requestHeader.getTopic())) {
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                    + "] sending message is forbidden");
            return response;
        }
        //校验消息主题的合法性
        TopicValidator.ValidateTopicResult result = TopicValidator.validateTopic(requestHeader.getTopic());
        if (!result.isValid()) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(result.getRemark());
            return response;
        }//校验消息主题是否为系统内置的一些不允许发送消息的主题，如果是则直接返回错误响应
        if (TopicValidator.isNotAllowedSendTopic(requestHeader.getTopic())) {
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark("Sending message to topic[" + requestHeader.getTopic() + "] is forbidden.");
            return response;
        }
        //从配置管理器中得到对应主题的配置信息
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
        if (null == topicConfig) {
            //如果当前broker节点中没有对应的主题配置，那就可以自动创建主题
            int topicSysFlag = 0;
            if (requestHeader.isUnitMode()) {
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
                } else {
                    topicSysFlag = TopicSysFlag.buildSysFlag(true, false);
                }
            }

            LOGGER.warn("the topic {} not exist, producer: {}", requestHeader.getTopic(), ctx.channel().remoteAddress());
            //下面就是创建对应主题信息的操作
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(
                    requestHeader.getTopic(),
                    requestHeader.getDefaultTopic(),
                    RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                    requestHeader.getDefaultTopicQueueNums(), topicSysFlag);

            if (null == topicConfig) {
                //没有创建成功则重新创建
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    topicConfig =
                            this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
                                    requestHeader.getTopic(), 1, PermName.PERM_WRITE | PermName.PERM_READ,
                                    topicSysFlag);
                }
            }

            if (null == topicConfig) {
                //重试之后仍然没有创建成功，则返回失败响应
                response.setCode(ResponseCode.TOPIC_NOT_EXIST);
                response.setRemark("topic[" + requestHeader.getTopic() + "] not exist, apply first please!"
                        + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
                return response;
            }
        }
        //执行到这里意味着已经获取到了主题的配置信息，接下来就要得到消息发送的目标队列
        int queueIdInt = requestHeader.getQueueId();
        //计算主题中的最大数量队列，也就是取读队列和写队列的最大值
        int idValid = Math.max(topicConfig.getWriteQueueNums(), topicConfig.getReadQueueNums());
        //然后判断目标队列的Id是否大于最大的队列Id
        if (queueIdInt >= idValid) {
            //如果大于则返回错误响应
            String errorInfo = String.format("request queueId[%d] is illegal, %s Producer: %s",
                    queueIdInt,
                    topicConfig,
                    RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

            LOGGER.warn(errorInfo);
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(errorInfo);

            return response;
        }
        return response;
    }

    public void registerSendMessageHook(List<SendMessageHook> sendMessageHookList) {
        this.sendMessageHookList = sendMessageHookList;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建发送消息上下文对象的方法
     */
    protected SendMessageContext buildMsgContext(ChannelHandlerContext ctx, SendMessageRequestHeader requestHeader, RemotingCommand request) {
        //从请求的主题中提取命名空间，命名空间用于资源隔离
        String namespace = NamespaceUtil.getNamespaceFromResource(requestHeader.getTopic());
        SendMessageContext sendMessageContext;
        //创建SendMessageContext对象
        sendMessageContext = new SendMessageContext();
        //设置命名空间，用于多租户隔离
        sendMessageContext.setNamespace(namespace);
        //设置生产者组
        sendMessageContext.setProducerGroup(requestHeader.getProducerGroup());
        //设置消息主题
        sendMessageContext.setTopic(requestHeader.getTopic());
        //设置消息体字节大小
        sendMessageContext.setBodyLength(request.getBody().length);
        //设置消息的属性
        sendMessageContext.setMsgProps(requestHeader.getProperties());
        //设置消息产生的主机地址
        sendMessageContext.setBornHost(RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
        //设置当前Broker节点地址
        sendMessageContext.setBrokerAddr(this.brokerController.getBrokerAddr());
        //设置目标消息队列Id
        sendMessageContext.setQueueId(requestHeader.getQueueId());
        //设置当前Broker节点所在区域Id
        sendMessageContext.setBrokerRegionId(this.brokerController.getBrokerConfig().getRegionId());
        //设置消息生成时间戳
        sendMessageContext.setBornTimeStamp(requestHeader.getBornTimestamp());
        //设置当前请求对象生成的时间戳
        sendMessageContext.setRequestTimeStamp(System.currentTimeMillis());
        //从请求的扩展字段中得到商业用户信息
        String owner = request.getExtFields().get(BrokerStatsManager.COMMERCIAL_OWNER);
        //设置商业用户信息
        sendMessageContext.setCommercialOwner(owner);
        //将消息属性字符串转换为Map对象
        Map<String, String> properties = MessageDecoder.string2messageProperties(requestHeader.getProperties());
        //获取消息的唯一Id
        String uniqueKey = properties.get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
        //添加区域Id属性，标识消息产生的区域
        properties.put(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
        //添加跟踪开关属性，决定是否开启消息轨迹
        properties.put(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
        //将更新后的属性Map对象转换回字符串，更新到请求头中
        requestHeader.setProperties(MessageDecoder.messageProperties2String(properties));
        //如果消息唯一Id为空，设置为空字符串
        if (uniqueKey == null) {
            uniqueKey = "";
        }
        //设置消息唯一Id到上下文对象中
        sendMessageContext.setMsgUniqueKey(uniqueKey);
        //判断消息类型：如果包含分片键属性，也就是生产者组信息，那当前消息就是顺序消息，否则就是普通消息
        if (properties.containsKey(MessageConst.PROPERTY_SHARDING_KEY)) {
            //设置消息类型为顺序消息
            sendMessageContext.setMsgType(MessageType.Order_Msg);
        } else {
            //设置消息类型为普通消息
            sendMessageContext.setMsgType(MessageType.Normal_Msg);
        }
        //返回构建好的发送消息上下文对象
        return sendMessageContext;
    }


    public SocketAddress getStoreHost() {
        return brokerController.getStoreHost();
    }
    public boolean hasSendMessageHook() {
        return sendMessageHookList != null && !this.sendMessageHookList.isEmpty();
    }


    public void executeSendMessageHookBefore(SendMessageContext context) {
        if (hasSendMessageHook()) {
            for (SendMessageHook hook : this.sendMessageHookList) {
                try {
                    hook.sendMessageBefore(context);
                } catch (AbortProcessException e) {
                    throw e;
                } catch (Throwable e) {
                    //ignore
                }
            }
        }
    }

    protected void doResponse(ChannelHandlerContext ctx, RemotingCommand request, final RemotingCommand response) {
        NettyRemotingAbstract.writeResponse(ctx.channel(), request, response);
    }


    protected int randomQueueId(int writeQueueNums) {
        return ThreadLocalRandom.current().nextInt(99999999) % writeQueueNums;
    }

    protected SendMessageRequestHeader parseRequestHeader(RemotingCommand request) throws RemotingCommandException {
        return SendMessageRequestHeader.parseRequestHeader(request);
    }

    public void executeSendMessageHookAfter(final RemotingCommand response, final SendMessageContext context) {
        if (hasSendMessageHook()) {
            for (SendMessageHook hook : this.sendMessageHookList) {
                try {
                    if (response != null) {
                        final SendMessageResponseHeader responseHeader =
                                (SendMessageResponseHeader) response.readCustomHeader();
                        context.setMsgId(responseHeader.getMsgId());
                        context.setQueueId(responseHeader.getQueueId());
                        context.setQueueOffset(responseHeader.getQueueOffset());
                        context.setCode(response.getCode());
                        context.setErrorMsg(response.getRemark());
                    }
                    hook.sendMessageAfter(context);
                } catch (Throwable e) {
                    //ignore
                }
            }
        }
    }

    @Override
    public boolean rejectRequest() {
        return false;
    }
}
